#include <block_storage.h>
#include <zookeeper.h>
#include <stdlib.h>
#include <time.h>

pthread_mutex_t lock;
pthread_cond_t cond;
int total_count;
int size;

long get_time();
void init(int count){
    pthread_mutex_init(&lock,0);
    pthread_cond_init(&cond,0);
    total_count = count;
}

void wait_for_finish(){
    pthread_mutex_lock(&lock);
    while(total_count>0){
	pthread_cond_wait(&cond, &lock);
    }
    pthread_mutex_unlock(&lock);
}

int requestsInInterval = 0;
long lastOutputTime = 0;
long startTime = 0;
void finish(){
    pthread_mutex_lock(&lock);
    total_count--;
    //printf("count %d\n", total_count);
    if(total_count<=0){
        pthread_cond_broadcast(&cond);
    }
    //if(startTime!=0)
    //    printf("%ld\n", get_time()-startTime);
    /*requestsInInterval++;
    long gap = get_time() - lastOutputTime;
    if(gap >= 1000){
	printf("%ld %f\n", gap, (float)requestsInInterval*1000.0/(float)gap);
	requestsInInterval = 0;
	lastOutputTime = get_time();
    }*/
    pthread_mutex_unlock(&lock);
}


void write_completion(int rc, const void *data){
    finish();
}

void read_completion(int rc, const char *value, int value_len,
        const void *data){
    //printf("read size = %d\n", value_len);
    if(value_len!=size)
	printf("size unmatch %d %d\n", size, value_len);
    finish();
}


int main(int argc, char **argv){
    if(argc!=7){
	printf("test <id> <totalClients> <size> <count> <read/write> <sequential/random>\n");
	return -1;
    }
    int id = atoi(argv[1]);
    int totalClients = atoi(argv[2]);

    srand(id+time(NULL));
    block_storage_t *storage = block_storage_init("application1:2181,application2:2181,application3:2181", id);
    size = atoi(argv[3]);
    int count = atoi(argv[4]);
    int read_ops = atoi(argv[5]);
    int sequential = atoi(argv[6]);
    
    char *buf = malloc(size);
    memset(buf, 0, size);
    //int no_of_blocks = 96*1024*1024/size*1024;
    int no_of_blocks = 96*1024*1024/size*1024;
    int startOff = rand()%no_of_blocks;
    startOff = no_of_blocks/totalClients*(id%totalClients);
    int i=0;
    sleep(5);
    for(i=0; i<10; i++)
        write_block(storage, i, buf, size, write_completion,0);
    sleep(15);
    init(count);
    long start = get_time();
    startTime = get_time();
    fprintf(stderr,"startOff = %d %ld\n", startOff,startTime);
    for(i=0; i<count; i++){
	int blockNo;
	if(sequential)
	    blockNo = (i+startOff)%no_of_blocks;//+(id-1)*no_of_blocks;
	else
	    blockNo = rand()%no_of_blocks;//+(id-1)*no_of_blocks;
	if(read_ops)
            read_block(storage, blockNo, read_completion, 0);
	else{
	    buf[0]=(char)i%128;
	    write_block(storage, blockNo, buf, size, write_completion,0);
	}
	if(i==count/5)
	    start = get_time();
    }
    wait_for_finish();
    printf("throughput = %f req/sec\n", count*0.8*1000.0/(get_time()-start));
}
